﻿
Imports System.IO
Imports std
Imports storoomc
Imports storcore

Public Interface slicebase
    Inherits comparableStorageInterface(Of slicebase)
    Function filename() As String
    Overloads Function open(ByVal filename As String) As Boolean
    Function create() As Boolean
    Function create(ByVal filename As String) As Boolean
End Interface

Partial Public MustInherit Class _slice(Of ATOM_T)
    Implements slicebase, IComparable(Of _slice(Of ATOM_T)), IComparable

    Protected MustOverride Function filter_slicetype_value() As String
    Protected MustOverride Function atom_ctor() As ATOM_T
    Protected MustOverride Function closeImpl() As Boolean
    Protected MustOverride Function deleteImpl(ByVal atom As ATOM_T) As Boolean
    Protected MustOverride Function modifyImpl(ByVal atom As ATOM_T, ByVal buff As bytes) As Boolean
    Protected MustOverride Function readImpl(ByVal atom As ATOM_T, ByRef buff As bytes) As Boolean
    Protected MustOverride Function appendImpl(ByVal atom As ATOM_T, ByVal buff As bytes) As Boolean
    Protected MustOverride Function writeImpl(ByVal key As String, ByVal atom As ATOM_T, ByVal buff As bytes) As Boolean
    Protected MustOverride Function retireImpl() As Boolean
    Protected MustOverride Function sizeofImpl(ByVal atom As ATOM_T, ByRef size As UInt64) As Boolean
    Protected MustOverride Function version() As String
    Protected MustOverride Function maxSupportVersion() As String
    Protected MustOverride Function minSupportVersion() As String
    Protected MustOverride Function readIndex(ByRef key As String, ByRef atom As ATOM_T, _
                                              ByVal lastpos As UInt64) As Boolean
End Class

Partial Public MustInherit Class _slice(Of ATOM_T)
    Implements slicebase, IComparable(Of _slice(Of ATOM_T)), IComparable

    Private Const NO_STREAM_BUFFER As Int32 = 8
    Private _datapath As String = Nothing
    Private _filename As String = Nothing
    Private _stream As Stream = Nothing
    Private _reader As BinaryReader = Nothing
    Private _writer As BinaryWriter = Nothing
    Private lock As sequentialLock = Nothing
    Private atoms As map(Of String, ATOM_T) = Nothing
    Private atomcount As Int64 = 0
    Private _autodropDirtyData As Boolean = False
    Private _filters As map(Of String, String) = Nothing
    Private memoryBuffSize As Int32 = npos

    Protected Function stream() As Stream
        Return _stream
    End Function

    Protected Function reader() As BinaryReader
        Return _reader
    End Function

    Protected Function writer() As BinaryWriter
        Return _writer
    End Function

    Public Function filename() As String Implements slicebase.filename
        assert(Not isemptystring(_filename), "_filename has not been set.")
        assert(Not Path.IsPathRooted(_filename), "_filename is from root.")
        Static __saved_filename As String = Nothing
        If __saved_filename Is Nothing Then
            copy(__saved_filename, _filename)
        Else
            assert(strsame(__saved_filename, _filename), "should not change filename after create / open.")
        End If
        Return _filename
    End Function

    Public Function datapath() As String
        Return _datapath
    End Function

    Protected Function autodropDirtyData() As Boolean
        Return _autodropDirtyData
    End Function

    Private Function find_atom(ByVal key As String, ByRef atom As ATOM_T, _
                               Optional ByRef it As map(Of String, ATOM_T).iterator = Nothing) As Boolean
        it = atoms.find(key)
        If it = atoms.end() Then
            Return False
        Else
            atom = (+it).second
            Return True
        End If
    End Function

    Public Function capacity(ByRef cap As UInt64) As Boolean Implements storageInterface.capacity
        assert(False, "do not support capacity check to a single slice.")
        Return False
    End Function

    Private Function _close() As Boolean
        If Not atoms Is Nothing Then
            atoms.clear()
        End If
        Dim rtn As Boolean
        rtn = closeImpl()
        closestream(_stream)
        closeBinaryReader(_reader)
        closeBinaryWriter(_writer)

        Return rtn
    End Function

    Public Function close() As Boolean Implements storageInterface.close
        lock.wait()
        Dim rtn As Boolean
        rtn = _close()
        lock.release()

        Return rtn
    End Function

    Private Function _delete(ByVal key As String) As Boolean
        Dim atom As ATOM_T = Nothing
        Dim it As map(Of String, ATOM_T).iterator = Nothing
        If find_atom(key, atom, it) Then
            Return startdiskoper() And (deleteImpl(atom) AndAlso atoms.erase(it)) And enddiskoper()
        Else
            Return False
        End If
    End Function

    Public Function delete(ByVal key As String) As Boolean Implements storageInterface.delete
        If key Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = _delete(key)
        lock.release()

        Return rtn
    End Function

    Public Function empty() As Boolean Implements storageInterface.empty
        Return atoms.empty()
    End Function

    Protected Overridable Function fullImpl() As Boolean
        Return atoms.size() >= atomcount
    End Function

    'since open will create stream / fceslite, so need to lock
    Public Function full() As Boolean Implements storageInterface.full
        lock.wait()
        Dim rtn As Boolean
        rtn = fullImpl()
        lock.release()

        Return rtn
    End Function

    Private Sub heartbeat()
        If _open() AndAlso _health(False) Then
            raiseError("return normal after re-open " + filename(), errorHandle.errorType.application)
        Else
            raiseError("still abnormal after re-open " + filename(), errorHandle.errorType.exclamation)
        End If
    End Sub

    Protected Overridable Function healthImpl(ByVal needDetailedMessage As Boolean) As Boolean
        Return True
    End Function

    Private Function _health(Optional ByVal needDetailedMessage As Boolean = True) As Boolean
        If atoms Is Nothing Then
            If needDetailedMessage Then
                raiseError("atoms is nothing when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If isemptystring(filename()) Then
            If needDetailedMessage Then
                raiseError("filename is empty string when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If stream() Is Nothing Then
            If needDetailedMessage Then
                raiseError("stream is nothing when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If Not stream().CanRead Then
            If needDetailedMessage Then
                raiseError("stream cannot read when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If Not stream().CanWrite Then
            If needDetailedMessage Then
                raiseError("stream cannot write when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If Not stream().CanSeek Then
            If needDetailedMessage Then
                raiseError("stream cannot seek when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If reader() Is Nothing Then
            If needDetailedMessage Then
                raiseError("reader is nothing when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        If writer() Is Nothing Then
            If needDetailedMessage Then
                raiseError("writer is nothing when check health.", errorHandle.errorType.warning)
            End If
            Return False
        End If

        Return healthImpl(needDetailedMessage)
    End Function

    'to avoid open twice, so lock it
    Public Function health() As Boolean Implements storageInterface.health
        Dim rtn As Boolean
        lock.wait()
        If Not _health(False) Then
            heartbeat()
        End If
        rtn = _health(True)
        lock.release()

        Return rtn
    End Function

    Public Function keycount(ByRef count As UInt64) As Boolean Implements storageInterface.keycount
        count = atoms.size()
        Return True
    End Function

    Public Function list(ByRef keys As vector(Of String)) As Boolean Implements storageInterface.list
        If keys Is Nothing Then
            keys = New vector(Of String)()
        Else
            keys.clear()
        End If

        Dim it As map(Of String, ATOM_T).iterator = Nothing
        lock.wait()
        it = atoms.begin()
        While it <> atoms.end()
            keys.push_back((+it).first())
            it += 1
        End While
        lock.release()

        Return True
    End Function

    Private Function _write(ByVal key As String, ByVal buff As bytes) As Boolean
        Dim atom As ATOM_T = Nothing
        atom = atom_ctor()
        atoms.insert(key, atom)
        Return startdiskoper() And writeImpl(key, atom, buff) And enddiskoper()
    End Function

    Private Function _modify(ByVal key As String, ByVal buff As bytes) As Boolean
        Dim atom As ATOM_T = Nothing
        If find_atom(key, atom) Then
            Return startdiskoper() And modifyImpl(atom, buff) And enddiskoper()
        Else
            Return _write(key, buff)
        End If
    End Function

    Public Function modify(ByVal key As String, ByVal buff As bytes) As Boolean Implements storageInterface.modify
        If key Is Nothing OrElse buff Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = _modify(key, buff)
        lock.release()

        Return rtn
    End Function

    Protected Function create_empty_slice() As Boolean
        If stream() Is Nothing OrElse writer() Is Nothing Then
            If Not _open_stream() Then
                raiseError(errorHandle.errorType.exclamation, "cannot open file stream, filesystem error.")
                Return False
            End If
        End If
        Try
            stream().SetLength(0)
            stream().Flush()
            writer().Seek(0, SeekOrigin.Begin)
            writer().Write(version())
            writer().Flush()
            Return True
        Catch ex As Exception
            raiseError("cannot create empty slice, ex " + ex.Message, errorHandle.errorType.exclamation)
            Return False
        End Try
    End Function

    Protected Overridable Sub create_start()
    End Sub

    Private Function createImpl() As Boolean
        If File.Exists(fullfilepath()) Then
            Return False
        Else
            'make sure the file should be existing
            File.WriteAllText(fullfilepath(), emptyString)
            Dim rtn As Boolean
            rtn = startdiskoper() And (_open_stream() AndAlso create_empty_slice()) And enddiskoper()
            create_start()
            Return True
        End If
    End Function

    Public Function create(ByVal filename As String) As Boolean Implements slicebase.create
        lock.wait()
        copy(_filename, filename)
        Dim rtn As Boolean
        rtn = createImpl()
        lock.release()

        Return rtn
    End Function

    Public Function create() As Boolean Implements slicebase.create
        lock.wait()
        _filename = createGUID()
        While File.Exists(fullfilepath())
            raiseError("congratulations, the guid points to an existing file ... try to set another filename.")
            _filename = createGUID()
        End While
        Dim rtn As Boolean
        rtn = createImpl()
        lock.release()

        Return rtn
    End Function

    Private Function fullfilepath() As String
        Return diskslices.datafile(datapath(), filename())
    End Function

    Private Function check_version() As Boolean
        stream().Seek(0, SeekOrigin.Begin)
        Dim s As String = Nothing
        If Not binaryReaderHelper.readString(reader(), s) Then
            raiseError("cannot get version info from file " + fullfilepath() + _
                       ", invalid or not supported slice file.", errorHandle.errorType.exclamation)
            _close()
            Return False
        End If

        kickLeftRightBlank(s)
        If s <= maxSupportVersion() AndAlso s >= minSupportVersion() Then
            Return True
        Else
            raiseError("not supported version " + s, errorHandle.errorType.exclamation)
            _close()
            Return False
        End If
    End Function

    Private Function _open_stream() As Boolean
        closeBinaryWriter(_writer)
        closeBinaryReader(_reader)
        closestream(_stream)
        Try
            _stream = New FileStream(fullfilepath(), FileMode.Open, FileAccess.ReadWrite, FileShare.Read, _
                                     memoryBuffSize)
            _reader = New BinaryReader(_stream, encoding.utf8())
            _writer = New BinaryWriter(_stream, encoding.utf8())
            Return True
        Catch ex As Exception
            raiseError("cannot create stream / reader / writer, ex " + ex.Message, errorHandle.errorType.exclamation)
            Return False
        End Try
    End Function

    Private Function _open() As Boolean
        If _open_stream() Then
            startdiskoper()
            If check_version() Then
                Dim lastpos As UInt64
                lastpos = stream().Position()
                Dim key As String = Nothing
                Dim atom As ATOM_T = Nothing
                While readIndex(key, atom, lastpos)
                    If stream().Position() > stream().Length() Then
                        raiseError(errorHandle.errorType.exclamation, _
                                   "stream().Position() > stream().Length(), pos ", stream().Position(), _
                                   ", len ", stream().Length(), ", assume data incorrect, and ignore.")
                        Exit While
                    End If
                    If atoms.find(key) = atoms.end() Then
                        atoms.insert(key, atom)
                    Else
                        raiseError("found an existing key from atoms, filename " + filename() + ", key " + key, _
                                   errorHandle.errorType.exclamation)
                    End If
                    lastpos = stream().Position
                End While

                assert(lastpos <= stream().Length())
                If lastpos < stream().Length() Then
                    raiseError("not enough data from file " + filename() + _
                               ", invalid index file, ignore following data.", errorHandle.errorType.exclamation)
                    Try
                        stream().SetLength(lastpos)
                    Catch ex As Exception
                        raiseError(errorHandle.errorType.exclamation, _
                                   "cannot set stream length to ", lastpos, ", ex ", ex.Message)
                    End Try
                End If
            End If
            enddiskoper()
            open_start()
            Return True
        Else
            Return False
        End If
    End Function

    Protected Overridable Sub open_start()
    End Sub

    Public Function open() As Boolean Implements storageInterface.open
        lock.wait()
        Dim rtn As Boolean
        rtn = _open()
        lock.release()

        Return rtn
    End Function

    Public Function open(ByVal filename As String) As Boolean Implements slicebase.open
        lock.wait()
        Dim rtn As Boolean
        copy(_filename, filename)
        rtn = _open()
        lock.release()

        Return rtn
    End Function

    Private Function _read(ByVal key As String, ByRef buff As bytes) As Boolean
        Dim atom As ATOM_T = Nothing
        If find_atom(key, atom) Then
            Return startdiskoper() And readImpl(atom, buff) And enddiskoper()
        Else
            Return False
        End If
    End Function

    Public Function read(ByVal key As String, ByRef buff As bytes) As Boolean Implements storageInterface.read
        If key Is Nothing Then
            Return False
        End If
        If buff Is Nothing Then
            buff = New bytes()
        Else
            buff.clear()
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = _read(key, buff)
        lock.release()
        Return rtn
    End Function

    Public Function retire() As Boolean Implements storageInterface.retire
        lock.wait()
        atoms.clear()
        Dim rtn As Boolean
        rtn = startdiskoper() And retireImpl() And enddiskoper()
        lock.release()

        Return rtn
    End Function

    Public Function seek(ByVal key As String) As Boolean Implements storageInterface.seek
        If key Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = find_atom(key, Nothing)
        lock.release()

        Return rtn
    End Function

    Private Function sizeofImpl(ByVal key As String, ByRef size As UInt64) As Boolean
        Dim atom As ATOM_T = Nothing
        If find_atom(key, atom) Then
            Return sizeofImpl(atom, size)
        Else
            Return False
        End If
    End Function

    Public Function sizeof(ByVal key As String, ByRef size As UInt64) As Boolean Implements storageInterface.sizeof
        If key Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = sizeofImpl(key, size)
        lock.release()

        Return rtn
    End Function

    Public Function status(ByRef allSlicesCount As UInt64, ByRef unhealthSlicesCount As UInt64) As Boolean _
        Implements storageInterface.status
        assert(False, "do not support status check to single slice.")
        Return False
    End Function

    Public Function sync(ByVal desthost As String, ByVal destport As Int32) As Boolean Implements storageInterface.sync
        Return synchelper(Me, desthost, destport)
    End Function

    Public Function uniqueWrite(ByVal key As String, ByVal buff As bytes) As Boolean _
        Implements storageInterface.uniqueWrite
        If key Is Nothing OrElse buff Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        rtn = Not find_atom(key, Nothing) AndAlso _write(key, buff)
        lock.release()

        Return rtn
    End Function

    Public Function valuesize(ByRef size As UInt64) As Boolean Implements storageInterface.valuesize
        size = 0
        lock.wait()
        Dim it As map(Of String, ATOM_T).iterator = Nothing
        it = atoms.begin()
        While it <> atoms.end()
            Dim this As UInt64
            If Not sizeofImpl((+it).second, this) Then
                Exit While
            Else
                size += this
            End If
            it += 1
        End While

        Dim rtn As Boolean
        rtn = (it = atoms.end())
        lock.release()

        Return rtn
    End Function

    Public Function write(ByVal key As String, ByVal buff As bytes) As Boolean Implements storageInterface.write
        If key Is Nothing OrElse buff Is Nothing Then
            Return False
        End If

        lock.wait()
        Dim rtn As Boolean
        Dim atom As ATOM_T = Nothing
        If find_atom(key, atom) Then
            rtn = startdiskoper() And appendImpl(atom, buff) And enddiskoper()
        Else
            rtn = _write(key, buff)
        End If
        lock.release()

        Return rtn
    End Function

    Public Function CompareTo(ByVal other As slicebase) As Int32 Implements IComparable(Of slicebase).CompareTo
        Return cast_compare(Of _slice(Of ATOM_T), slicebase, _slice(Of ATOM_T))(Me, other)
    End Function

    Public Function CompareTo(ByVal obj As Object) As Int32 Implements IComparable.CompareTo
        Return cast_compare(Of _slice(Of ATOM_T), Object, _slice(Of ATOM_T))(Me, obj)
    End Function

    Public Function CompareTo(ByVal other As _slice(Of ATOM_T)) As Int32 _
        Implements IComparable(Of _slice(Of ATOM_T)).CompareTo
        Dim cmp As Int32
        cmp = object_compare(Me, other)
        If cmp = object_compare_undetermined Then
            Return compare(Me.filename(), other.filename())
        Else
            Return cmp
        End If
    End Function

    Protected Function filters() As map(Of String, String)
        If _filters Is Nothing Then
            _filters = New map(Of String, String)()
            _filters(configNaming.filter.slicetype) = filter_slicetype_value()
        End If

        Return _filters
    End Function

    Protected Delegate Function traversal_atoms_action(ByVal atom As ATOM_T) As Boolean

    Protected Function traversal_atoms(ByVal action As traversal_atoms_action) As Boolean
        Dim it As map(Of String, ATOM_T).iterator = Nothing
        it = atoms.begin()
        While it <> atoms.end()
            If Not action((+it).second) Then
                Exit While
            End If
            it += 1
        End While

        Return it = atoms.end()
    End Function

    Protected Function lock_traversal_atoms(ByVal action As traversal_atoms_action) As Boolean
        Dim rtn As Boolean
        lock.wait()
        rtn = traversal_atoms(action)
        lock.release()
        Return rtn
    End Function

    Protected Function read_slicesize() As UInt64
        Return config.parameter(configNaming.stornode.name, _
                                configNaming.stornode.slicesize, _
                                configNaming.stornode.defaultValue.slicesize, _
                                filters())
    End Function

    Public Sub New()
        lock = New sequentialLock()
        atoms = New map(Of String, ATOM_T)()
        atomcount = config.parameter(configNaming.stornode.name, _
                                     configNaming.stornode.atomcount, _
                                     configNaming.stornode.defaultValue.atomcount, _
                                     filters())
        _autodropDirtyData = config.parameter(configNaming.stornode.name, _
                                              configNaming.stornode.autodropDirtyData, _
                                              configNaming.stornode.defaultValue.autodropDirtyData, _
                                              filters())
        'it must be consistent with the one in diskslices
        copy(_datapath, stornode.datapath())
        Dim s As String = Nothing
        If Not config.getparameter(configNaming.stornode.name, _
                                   configNaming.stornode.memoryBuffSize, s, _
                                   configNaming.stornode.defaultValue.memoryBuffSize, _
                                   filters()) OrElse _
           Not convertor.toint32(s, memoryBuffSize) Then
            memoryBuffSize = (availablePhysicalMemory() >> 14)
            'for 2G memory, use 131072 bytes as buffer, so can contain about 16384 slices
        ElseIf memoryBuffSize <= NO_STREAM_BUFFER Then
            memoryBuffSize = NO_STREAM_BUFFER
        End If
    End Sub

    'help functions for slice disk oper analysis
    Private Shared STORNODE_SLICE_DISK_OPER_TIMETICKS As Int64

    Shared Sub New()
        STORNODE_SLICE_DISK_OPER_TIMETICKS = _
            counter.registerAverageAndLastAverage("STORNODE_SLICE_DISK_OPER_TIMETICKS")
    End Sub

    Private Class startticks
        <ThreadStatic()> Public Shared t As Int64
    End Class

    Protected Shared Function startdiskoper() As Boolean
        startticks.t = Now.Ticks()
        Return True
    End Function

    Protected Overridable Sub flush()
        writer().Flush()
        stream().Flush()
    End Sub

    Protected Function enddiskoper() As Boolean
        If memoryBuffSize = NO_STREAM_BUFFER AndAlso (Not (writer() Is Nothing OrElse stream() Is Nothing)) Then
            flush()
        End If
        'since disk oper is the basic operation for stornode, so sample 1% to count is enough
        counter.increase(STORNODE_SLICE_DISK_OPER_TIMETICKS, Now.Ticks() - startticks.t)

        Return True
    End Function
End Class